package com.bomb.disruptor.db;

import com.bomb.disruptor.base.LongEventFactory;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.InputStreamReader;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class InsertSqlMain {

    static String sqlFilePath = "C:\\Users\\Waferslove\\Desktop\\dm.sql";//sql文件的路径

    public static void main(String[] args) throws Exception {
        //创建缓冲池
        ExecutorService executor = Executors.newFixedThreadPool(64);
        //创建工厂
        EventFactory<Sql> factory = new EventFactory<Sql>() {
            public Sql newInstance() {
                return new Sql();
            }
        };        //创建bufferSize ,也就是RingBuffer大小，必须是2的N次方
        int ringBufferSize = 1024 * 1024; //

        // 创建disruptor
        Disruptor<Sql> disruptor =
                new Disruptor<Sql>(factory, ringBufferSize, executor, ProducerType.SINGLE, new YieldingWaitStrategy());
        // 连接消费事件方法

        InsertHandler[] handlers = new InsertHandler[32];
        for (int i = 0; i < handlers.length; i++) {
            handlers[i] = new InsertHandler("insert" + i);
        }

        disruptor.handleEventsWith(handlers);

        // 启动
        disruptor.start();

        // Disruptor 的事件发布过程是一个两阶段提交的过程：
        // 发布事件，环形数据结构
        RingBuffer<Sql> ringBuffer = disruptor.getRingBuffer();

        InsertProducer producer = new InsertProducer(ringBuffer);


        //下面这个生产8个数据
        File file = new File(sqlFilePath);
        BufferedReader br = new BufferedReader(new InputStreamReader(new FileInputStream(file), "UTF-8"));
        //替换
        String line = null;
        //以行为单位进行遍历
        long index = 0;
        while ((line = br.readLine()) != null) {
            index++;
            producer.onData(line, index);
            System.out.println("丢入：" + index);
            if (index % 10000 == 0) {
                Thread.sleep(1000);
            }
        }
        System.out.println("---------------开始执行-----------------");

        disruptor.shutdown();//关闭 disruptor，方法会堵塞，直至所有的事件都得到处理；
        executor.shutdown();//关闭 disruptor 使用的线程池；如果需要的话，必须手动关闭， disruptor 在 shutdown 时不会自动关闭；


    }
}  